// Copyright 2015 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package raft

import (
	"errors"
	"github.com/pingcap-incubator/tinykv/log"
	pb "github.com/pingcap-incubator/tinykv/proto/pkg/eraftpb"
	"math/rand"
	"sync"
)

// None is a placeholder node ID used when there is no leader.
const None uint64 = 0

// StateType represents the role of a node in a cluster.
type StateType uint64

const (
	StateFollower StateType = iota
	StateCandidate
	StateLeader
)

var stmap = [...]string{
	"StateFollower",
	"StateCandidate",
	"StateLeader",
}

func (st StateType) String() string {
	return stmap[uint64(st)]
}

// ErrProposalDropped is returned when the proposal is ignored by some cases,
// so that the proposer can be notified and fail fast.
var ErrProposalDropped = errors.New("raft proposal dropped")

// Config contains the parameters to start a raft.
type Config struct {
	// ID is the identity of the local raft. ID cannot be 0.
	ID uint64

	// peers contains the IDs of all nodes (including self) in the raft cluster. It
	// should only be set when starting a new raft cluster. Restarting raft from
	// previous configuration will panic if peers is set. peer is private and only
	// used for testing right now.
	peers []uint64

	// ElectionTick is the number of Node.Tick invocations that must pass between
	// elections. That is, if a follower does not receive any message from the
	// leader of current term before ElectionTick has elapsed, it will become
	// candidate and start an election. ElectionTick must be greater than
	// HeartbeatTick. We suggest ElectionTick = 10 * HeartbeatTick to avoid
	// unnecessary leader switching.
	ElectionTick int
	// HeartbeatTick is the number of Node.Tick invocations that must pass between
	// heartbeats. That is, a leader sends heartbeat messages to maintain its
	// leadership every HeartbeatTick ticks.
	HeartbeatTick int

	// Storage is the storage for raft. raft generates entries and states to be
	// stored in storage. raft reads the persisted entries and states out of
	// Storage when it needs. raft reads out the previous state and configuration
	// out of storage when restarting.
	Storage Storage
	// Applied is the last applied index. It should only be set when restarting
	// raft. raft will not return entries to the application smaller or equal to
	// Applied. If Applied is unset when restarting, raft might return previous
	// applied entries. This is a very application dependent configuration.
	Applied uint64
}

func (c *Config) validate() error {
	if c.ID == None {
		return errors.New("cannot use none as id")
	}

	if c.HeartbeatTick <= 0 {
		return errors.New("heartbeat tick must be greater than 0")
	}

	if c.ElectionTick <= c.HeartbeatTick {
		return errors.New("election tick must be greater than heartbeat tick")
	}

	if c.Storage == nil {
		return errors.New("storage cannot be nil")
	}

	return nil
}

// Progress represents a follower’s progress in the view of the leader. Leader maintains
// progresses of all followers, and sends entries to the follower based on its progress.
type Progress struct {
	Match, Next uint64
}

type Raft struct {
	id uint64

	Term uint64
	Vote uint64

	// the log
	RaftLog *RaftLog

	// log replication progress of each peers
	Prs map[uint64]*Progress

	// this peer's role
	State StateType

	// votes records
	votes     map[uint64]bool
	voteFails map[uint64]bool

	// msgs need to send
	msgs []pb.Message

	// the leader id
	Lead uint64

	// heartbeat interval, should send
	heartbeatTimeout int

	// number of ticks since it reached last heartbeatTimeout.
	// only leader keeps heartbeatElapsed.
	heartbeatElapsed int
	// Ticks since it reached last electionTimeout when it is leader or candidate.
	// Number of ticks since it reached last electionTimeout or received a

	// 记录时间、超时时间、配置传入的超时时间
	electionElapsed int
	electionTimeout int
	electionTick    int

	// leadTransferee is id of the leader transfer target when its value is not zero.
	// Follow the procedure defined in section 3.10 of Raft phd thesis.
	// (https://web.stanford.edu/~ouster/cgi-bin/papers/OngaroPhD.pdf)
	// (Used in 3A leader transfer)
	leadTransferee uint64

	// Only one conf change may be pending (in the log, but not yet
	// applied) at a time. This is enforced via PendingConfIndex, which
	// is set to a value >= the log index of the latest pending
	// configuration change (if any). Config changes are only allowed to
	// be proposed if the leader's applied index is greater than this
	// value.
	// (Used in 3A conf change)
	PendingConfIndex uint64

	mutex sync.Mutex
}

// newRaft return a raft peer with the given config
func newRaft(c *Config) *Raft {
	if err := c.validate(); err != nil {
		panic(err.Error())
	}

	initial, confState, _ := c.Storage.InitialState()

	if len(c.peers) == 0 {
		c.peers = confState.Nodes
	}

	// Your Code Here (2A).
	raftInstance := &Raft{}

	raftInstance.id = c.ID

	// raft角色组件
	raftInstance.State = StateFollower
	raftInstance.Term = initial.Term
	raftInstance.Lead = None

	raftInstance.Vote = initial.Vote
	raftInstance.votes = make(map[uint64]bool)
	raftInstance.voteFails = make(map[uint64]bool)

	// raft RPC请求
	raftInstance.msgs = make([]pb.Message, 0)

	raftInstance.heartbeatTimeout = c.HeartbeatTick
	raftInstance.electionTick = c.ElectionTick
	raftInstance.electionTimeout = c.ElectionTick + rand.Intn(c.ElectionTick)

	// raft日志组件
	raftLog := newLog(c.Storage)
	raftLog.setRaftIndex(raftInstance.id)
	raftInstance.RaftLog = raftLog
	raftInstance.Prs = make(map[uint64]*Progress)
	// raft: 领导者

	for _, peer := range c.peers {
		raftInstance.votes[peer] = false
		raftInstance.voteFails[peer] = false
		raftInstance.Prs[peer] = &Progress{Match: 0, Next: 1}
	}

	log.Debugf("raft init. id = %d, len = %d", raftInstance.id, len(raftInstance.Prs))
	return raftInstance
}

func (r *Raft) QueryUniqueId() uint64 {
	//r.mutex.Lock()
	//defer r.mutex.Unlock()
	return r.id
}

// sendAppend sends an append RPC with new entries (if any) and the
// current commit index to the given peer. Returns true if a message was sent.
func (r *Raft) sendAppend(to uint64) bool {
	// Your Code Here (2A).

	entries := make([]*pb.Entry, 0)

	targetEntries := r.RaftLog.nextSendEntries(r.Prs[to].Next)
	for i := range targetEntries {
		targetEntry := targetEntries[i]
		entries = append(entries, &targetEntry)
	}

	prevEntry := r.RaftLog.logAtIndex(r.Prs[to].Next - 1)

	m := pb.Message{
		MsgType: pb.MessageType_MsgAppend,
		From:    r.id,
		To:      to,
		Term:    r.Term,
		LogTerm: prevEntry.Term,      // 论文对应prevLogTerm ，即将处理的下一个数据的Term
		Index:   prevEntry.Index,     // 论文对应prevLogIndex，即将处理的下一个数据的Index
		Commit:  r.RaftLog.committed, // 论文对应领导者的commit_index（（也就是当前Raft）
		Entries: entries}
	r.appendRPC(m)
	log.Infof("[RaftAlgorithm][id = %d] append_entries rpc send from %d to %d", r.id, r.id, to)
	return true
}

// sendHeartbeat sends a heartbeat RPC to the given peer.
func (r *Raft) sendHeartbeat(to uint64) {
	// Your Code Here (2A).
	log.Debugf("hearbeat rpc send from %d to %d, term = %d", r.id, to, r.Term)
	m := pb.Message{From: r.id, To: to, Term: r.Term, MsgType: pb.MessageType_MsgHeartbeat}
	r.appendRPC(m)
}

func (r *Raft) sendRequestVote(to uint64) {
	// Your Code Here (2A).
	log.Debugf("request_vote rpc send from %d to %d, term = %d", r.id, to, r.Term)
	lastLogEntry := r.RaftLog.logAtIndex(r.RaftLog.LastIndex())
	m := pb.Message{
		MsgType: pb.MessageType_MsgRequestVote,
		From:    r.id,
		To:      to,
		Term:    r.Term,
		Index:   lastLogEntry.Index,
		LogTerm: lastLogEntry.Term,
	}
	r.appendRPC(m)
}

// tick advances the internal logical clock by a single tick.
func (r *Raft) tick() {
	// Your Code Here (2A).
	switch r.State {
	case StateFollower:
		r.electionElapsed++
		if r.electionElapsed >= r.electionTimeout {
			r.triggerElectionTimeoutEvent()
			r.electionElapsed = 0
			r.electionTimeout = r.electionTick + rand.Intn(r.electionTick)
		}
	case StateCandidate:
		r.electionElapsed++
		if r.electionElapsed >= r.electionTimeout {
			r.triggerElectionTimeoutEvent()
			r.electionElapsed = 0
			r.electionTimeout = r.electionTick + rand.Intn(r.electionTick)
		}
	case StateLeader:
		r.heartbeatElapsed++
		if r.heartbeatElapsed >= r.heartbeatTimeout {
			r.triggerAllHearBeatEvent()
			r.heartbeatElapsed = 0
		}
	}
}

// becomeFollower transform this peer's state to Follower
func (r *Raft) becomeFollower(term uint64, lead uint64) {
	// Your Code Here (2A).
	// 任期和状态
	if r.Term < term {
		// 任期重设后，将Vote清空
		r.Term = term
		r.Vote = None
	}
	r.Lead = lead
	r.State = StateFollower
	log.Infof("raft %d become Follower. Term = %d, Lead = %d", r.id, r.Term, r.Lead)
}

// becomeCandidate transform this peer's state to candidate
func (r *Raft) becomeCandidate() {
	// Your Code Here (2A).
	r.Term = r.Term + 1
	r.Lead = r.id
	r.State = StateCandidate
	r.Vote = r.id
	r.votes[r.id] = true
	r.voteFails[r.id] = false
	log.Infof("raft %d become Candidate. Term = %d, Lead = %d", r.id, r.Term, r.Lead)

}

// becomeLeader transform this peer's state to leader
func (r *Raft) becomeLeader() {
	// Your Code Here (2A).
	// NOTE: Leader should propose a noop entry on its term
	r.Lead = r.id
	r.Vote = r.id
	r.State = StateLeader

	log.Infof("raft %d become Leader. Term = %d", r.id, r.Term)

	// 成为Leader时，首先生成一次NoopEntry，并伺机向其他节点发送
	noopEntry := pb.Entry{Term: r.Term, Index: r.RaftLog.LastIndex() + 1}
	r.RaftLog.append(noopEntry)
	r.Prs[r.id].Match = r.RaftLog.LastIndex()
	r.Prs[r.id].Next = r.Prs[r.id].Match + 1
	if r.isSingleInstance() {
		r.RaftLog.commit(noopEntry.Index)
		return
	}
	r.triggerAllAppendEntriesEvent()
}

// Step the entrance of handle message, see `MessageType`
// on `eraftpb.proto` for what msgs should be handled
func (r *Raft) Step(m pb.Message) error {
	// Your Code Here (2A).
	// common process for state

	switch m.MsgType {
	case pb.MessageType_MsgHup:
		return r.handleMsgHub(m)
	case pb.MessageType_MsgPropose:
		return r.handleMsgPropose(m)
	case pb.MessageType_MsgAppend:
		return r.handleMsgAppend(m)
	case pb.MessageType_MsgAppendResponse:
		return r.handleAppendEntries(m)
	case pb.MessageType_MsgRequestVote:
		return r.handleMsgRequestVote(m)
	case pb.MessageType_MsgRequestVoteResponse:
		return r.handleMsgRequestVoteResponse(m)
	case pb.MessageType_MsgBeat:
		return r.handleMsgBeat(m)
	case pb.MessageType_MsgHeartbeat:
		return r.handleMgsHeartbeat(m)
	case pb.MessageType_MsgHeartbeatResponse:
		return r.handleMgsHeartbeatResponse(m)
	default:
		return errors.New("unknown message type")
	}
}

// handleSnapshot handle Snapshot RPC request
func (r *Raft) handleSnapshot(m pb.Message) {
	// Your Code Here (2C).
}

// addNode add a new node to raft group
func (r *Raft) addNode(id uint64) {
	// Your Code Here (3A).
}

// removeNode remove a node from raft group
func (r *Raft) removeNode(id uint64) {
	// Your Code Here (3A).
}

// ------------------------------ 不同消息类型处理 START -----------------------------------
func (r *Raft) handleMsgHub(m pb.Message) error {
	switch r.State {
	case StateFollower:
		r.triggerElectionTimeoutEvent()
	case StateCandidate:
		r.triggerElectionTimeoutEvent()
	default:
		// ignore
	}
	return nil
}

func (r *Raft) handleMsgRequestVote(m pb.Message) error {
	//r.mutex.Lock()
	//defer r.mutex.Unlock()

	if r.Term < m.Term {
		log.Infof("[RequestVoteRequest] raft %d become Follower. r.Term = %d, target.Term = %d", r.id, r.Term, m.Term)
		r.becomeFollower(m.Term, None)
	}

	var isReject bool
	if r.Term <= m.Term {
		if r.Vote == None {
			// compare more up-to-date
			lastLog := r.RaftLog.logAtIndex(r.RaftLog.LastIndex())
			log.Infof("raft %d judge vote for %d. last_log_term =%d, last_log_index = %d, logTerm = %d, logIndex = %d",
				r.id, m.From, lastLog.Term, lastLog.Index, m.LogTerm, m.Index)

			if lastLog.Term < m.LogTerm {
				isReject = false
				log.Infof("raft %d judge vote for %d, m.Term = %d. is_rejct = %t. Term smaller", r.id, m.From, m.Term, isReject)
			} else if lastLog.Term > m.LogTerm {
				isReject = true
				log.Infof("raft %d judge vote for %d, m.Term = %d. is_rejct = %t. Term larger", r.id, m.From, m.Term, isReject)
			} else if lastLog.Index <= m.Index {
				// 此时Term已相等
				isReject = false
				log.Infof("raft %d judge vote for %d, m.Term = %d. is_rejct = %t. Index smaller or equal", r.id, m.From, m.Term, isReject)
			} else {
				isReject = true
				log.Infof("raft %d judge vote for %d, m.Term = %d. is_rejct = %t. Index larger", r.id, m.From, m.Term, isReject)
			}

		} else if r.Vote == m.From {
			log.Infof("raft %d judge vote for %d. duplicated", r.id, m.From)
			isReject = false
		} else {
			log.Infof("raft %d judge vote for %d. already vote = %d", r.id, m.From, r.Vote)
			isReject = true
		}
	} else if r.Term > m.Term {
		// source is in lower Term
		isReject = true
		log.Infof("raft %d judge vote for %d. Term is over.", r.id, m.From)
	}

	if !isReject {
		// 不拒绝的情况下，要进行投票
		r.Vote = m.From
	}

	response := pb.Message{From: r.id, To: m.From, Term: r.Term, MsgType: pb.MessageType_MsgRequestVoteResponse, Reject: isReject}
	r.appendRPC(response)
	log.Infof("[RequestVoteRPC] response from %d to %d, reject = %t", r.id, m.From, isReject)
	return nil
}

func (r *Raft) handleMsgRequestVoteResponse(m pb.Message) error {
	//r.mutex.Lock()
	//defer r.mutex.Unlock()

	if r.Term < m.Term {
		log.Infof("[RequestVoteResponse] raft %d become Follower. r.Term = %d, target.Term = %d", r.id, r.Term, m.Term)
		r.becomeFollower(m.Term, None)
	}

	if m.Reject {
		r.voteFails[m.From] = true
	} else {
		// 未被拒绝，也就是投票给自己
		r.votes[m.From] = true
	}

	if r.State == StateLeader || r.State == StateFollower {
		// 此时，不需要触发票数统计
		return nil
	}

	var voteCount = 0
	for _, v := range r.votes {
		if v {
			voteCount++
		}
	}
	if voteCount > len(r.votes)/2 {
		// 如果投票数量达到多数派，那么变成领导者
		r.becomeLeader()
		return nil
	}

	var voteFailsCount = 0
	for _, v := range r.voteFails {
		if v {
			voteFailsCount++
		}
	}
	if voteFailsCount > len(r.voteFails)/2 {
		// 如果投票数量达到多数派，那么变成领导者
		log.Infof("[C->F]raft %d become Follower. r.Term = %d, target.Term = %d. %+v", r.id, r.Term, m.Term, r.voteFails)
		r.becomeFollower(r.Term, None)
		return nil
	}
	return nil
}

func (r *Raft) handleMsgPropose(m pb.Message) error {
	if r.State != StateLeader {
		log.Errorf("rpc msg_propose state fail. msg =%+v", m)
		return nil
	}
	log.Infof("rpc msg_propose from %d to %d. size = %d", m.From, m.To, len(m.Entries))
	for _, entry := range m.GetEntries() {
		targetEntry := pb.Entry{
			EntryType: entry.EntryType,
			Term:      r.Term,
			Index:     r.RaftLog.LastIndex() + 1,
			Data:      entry.Data,
		}

		r.RaftLog.append(targetEntry)
		r.Prs[r.id].Match = r.RaftLog.LastIndex()
		r.Prs[r.id].Next = r.Prs[r.id].Match + 1
		if r.isSingleInstance() {
			r.RaftLog.commit(r.RaftLog.LastIndex())
		}
	}
	r.triggerAllAppendEntriesEvent()
	return nil
}

func (r *Raft) handleMsgAppend(m pb.Message) error {
	returnMsg := pb.Message{
		MsgType: pb.MessageType_MsgAppendResponse,
		From:    r.id,
		To:      m.From,
		Term:    r.Term,
		Index:   r.RaftLog.LastIndex(),
		Reject:  false,
	}

	log.Infof("[RaftAlgorithm][id = %d] append_entries rpc receive from %d to %d", r.id, m.From, m.To)

	if r.Term < m.Term {
		r.becomeFollower(m.Term, m.From)
	} else if r.Term == m.Term {
		if r.State == StateLeader {
			returnMsg.Reject = true
			r.appendRPC(returnMsg)
			return nil
		} else {
			r.becomeFollower(m.Term, m.From)
		}
	} else {
		returnMsg.Reject = true
		r.appendRPC(returnMsg)
		return nil
	}

	prevIndex := m.Index
	prevTerm := m.LogTerm

	if prevIndex > r.RaftLog.LastIndex() {
		returnMsg.Reject = true
		r.appendRPC(returnMsg)
		return nil
	}

	prevEntry := r.RaftLog.logAtIndex(prevIndex)
	// assert prevEntry != nil
	if prevTerm != prevEntry.Term {
		returnMsg.Reject = true
		r.appendRPC(returnMsg)
		return nil
	}

	for _, entry := range m.GetEntries() {
		targetEntry := pb.Entry{
			EntryType: entry.EntryType,
			Term:      entry.Term,
			Index:     entry.Index,
			Data:      entry.Data,
		}
		r.RaftLog.append(targetEntry)
	}

	lastEntry := prevEntry
	if len(m.GetEntries()) != 0 {
		lastEntry = m.GetEntries()[len(m.GetEntries())-1]
	}

	if m.Commit > r.RaftLog.committed {
		// 参考论文Figure2 AppendEntries RPC
		if m.Commit >= lastEntry.Index {
			r.RaftLog.commit(lastEntry.Index)
		} else {
			r.RaftLog.commit(m.Commit)
		}
	}

	returnMsg.Index = r.RaftLog.LastIndex()

	log.Infof("[RaftAlgorithm][id = %d] append_entries rpc receive success. commit = %d, apply=%d", r.id, r.RaftLog.committed, r.RaftLog.applied)
	r.appendRPC(returnMsg)
	return nil
}

// handleAppendEntries handle AppendEntries RPC request
func (r *Raft) handleAppendEntries(m pb.Message) error {
	log.Infof("[RaftAlgorithm][id = %d] append_entries rpc response m = %+v", r.id, m)
	//r.mutex.Lock()
	//defer r.mutex.Unlock()
	if m.Reject {
		return nil
	}

	if r.Prs[m.From].Match >= m.Index {
		// duplicate request
		return nil
	}

	r.Prs[m.From].Next = m.Index + 1
	r.Prs[m.From].Match = m.Index

	logEntry := r.RaftLog.logAtIndex(m.Index)
	if r.Term == logEntry.Term {
		// raft only commits log entries from current terms by counting replicas
		count := 1
		for id, peer := range r.Prs {
			if r.id == id {
				continue
			}
			if peer.Match >= r.Prs[m.From].Match {
				count++
			}
		}

		if count > len(r.Prs)/2 {
			r.RaftLog.commit(m.Index)
			r.triggerAllAppendEntriesEvent()
		}
	}

	log.Infof("[RaftAlgorithm][id = %d] append_entries rpc response success. commit=%d,apply=%d", r.id, r.RaftLog.committed, r.RaftLog.applied)
	return nil
}

func (r *Raft) handleMsgBeat(m pb.Message) error {
	if r.State != StateLeader {
		return nil
	}
	r.triggerAllHearBeatEvent()
	return nil
}

// handleMgsHeartbeat handle Heartbeat RPC request
func (r *Raft) handleMgsHeartbeat(m pb.Message) error {
	// Your Code Here (2A).
	log.Infof("[RaftAlgorithm][id = %d] heartbeat rpc response from %d", r.id, m.From)
	if r.Term < m.Term {
		r.becomeFollower(m.Term, m.From)
	}

	r.electionElapsed = 0

	returnMsg := pb.Message{
		MsgType: pb.MessageType_MsgHeartbeatResponse,
		From:    r.id,
		To:      m.From,
		Term:    r.Term,
		Index:   r.RaftLog.LastIndex(),
	}
	r.appendRPC(returnMsg)
	return nil
}

func (r *Raft) handleMgsHeartbeatResponse(m pb.Message) error {
	// Your Code Here (2A).
	if r.RaftLog.LastIndex() > m.Index {
		r.sendAppend(m.From)
	}
	return nil
}

// ------------------------------ 不同消息类型处理 STOP -----------------------------------

// ------------------------------ 触发事件 START -----------------------------------

func (r *Raft) triggerElectionTimeoutEvent() {
	r.becomeCandidate()
	r.triggerAllRequestVoteEvent()
}

func (r *Raft) triggerAllHearBeatEvent() {
	for id, _ := range r.Prs {
		if r.id != id {
			r.sendHeartbeat(id)
		}
	}
}

func (r *Raft) triggerAllAppendEntriesEvent() {
	log.Infof("raft %d trigger AllApendEntriesEvent", r.id)
	for id, _ := range r.Prs {
		if r.id != id {
			r.sendAppend(id)
		}
	}
}

func (r *Raft) triggerAllRequestVoteEvent() {
	log.Infof("raft %d start an Election. Term = %d, peers = %d", r.id, r.Term, len(r.Prs))
	if r.isSingleInstance() {
		r.becomeLeader()
		return
	}

	for id, _ := range r.Prs {
		if r.id != id {
			r.voteFails[id] = false
			r.sendRequestVote(id)
		}
	}
}

// ------------------------------ 触发事件 START -----------------------------------

func (r *Raft) isSingleInstance() bool {
	return len(r.Prs) == 1
}

// ------------------------------ 消息传递管理 -----------------------------------

func (r *Raft) fetchAllRPCAndRemake() []pb.Message {
	//r.mutex.Lock()
	//defer r.mutex.Unlock()

	messages := r.msgs
	r.msgs = make([]pb.Message, 0)
	return messages
}

func (r *Raft) queryRPCLength() int {
	//r.mutex.Lock()
	//defer r.mutex.Unlock()

	return len(r.msgs)
}

func (r *Raft) appendRPC(m pb.Message) {
	//r.mutex.Lock()
	//defer r.mutex.Unlock()

	r.msgs = append(r.msgs, m)
}
